#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>

#include "perf.h"
#include "base.h"

//TestConf *tf;

SafeQueue*
safequeue_create()
{
    SafeQueue *q;

    q = (SafeQueue *)zz_malloc(sizeof(SafeQueue));
    if (NULL == q) {
        DERROR("malloc safe queue error\n");
        return NULL;
    }
    
    memset(q, 0x0, sizeof(SafeQueue));

    int ret = pthread_mutex_init(&q->lock, NULL);
    if (ret == -1) {
        DERROR("pthread_mutex_init error\n");
        return NULL;
    }
    
    ret = pthread_cond_init(&q->cond_empty, NULL);
    if (ret == -1) {
        DERROR("pthread_cond_init error\n");
        return NULL;
    }
    ret = pthread_cond_init(&q->cond_full, NULL);
    if (ret == -1) {
        DERROR("pthread_cond_init error\n");
        return NULL;
    }
    q->size = 0;

    return q;
}

void
safequeue_destroy(SafeQueue *q)
{
    QueueItem *item, *tmp;

    item = q->head;

    while (item) {
        tmp = item;
        item = item->next;
        zz_free(tmp);
    }
    
    pthread_mutex_destroy(&q->lock);
    pthread_cond_destroy(&q->cond_empty);
    pthread_cond_destroy(&q->cond_full);
    zz_free(q);

    return;
}

int 
safequeue_put(SafeQueue *q, char *jstr)
{
    int  ret;

    QueueItem *item = (QueueItem *)zz_malloc(sizeof(QueueItem));
    if (NULL == item) {
        DERROR("malloc QueueItem error!\n");
        exit(-1);
        return -1;
    }
    item->json = jstr;
    
    ret = pthread_mutex_lock(&q->lock);
    while (q->size == tf->qsize) {
        pthread_cond_signal(&q->cond_full);
        ret = pthread_cond_wait(&q->cond_empty, &q->lock);
    }
    item->next = q->head;
    q->head = item;
    q->size++;
    q->dcount++;
    pthread_cond_signal(&q->cond_full);

    ret = pthread_mutex_unlock(&q->lock);

    return 0;
}

QueueItem*
safequeue_get(SafeQueue *q)
{
    int ret;
    QueueItem *item;

    ret = pthread_mutex_lock(&q->lock);
    if (ret != 0) {
        char errbuf[1024];
        strerror_r(errno, errbuf, 1024);
        DERROR("pthread_cond_broadcast error: %s\n", errbuf);
        exit(-1);
    }
    
    while (q->size == 0) {
        if (q->is_over == 1) {
            printf("thread %d is over\n", pthread_self());
            pthread_mutex_unlock(&q->lock);
            return NULL;
        }
        pthread_cond_signal(&q->cond_empty);
        ret = pthread_cond_wait(&q->cond_full, &q->lock);
    }
    item = q->head;
    q->head = item->next;
    q->size--;
    pthread_mutex_unlock(&q->lock);

    return item;
}

